package com.jerry.hm.datastructure.blockingqueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单锁实现阻塞队列
 *
 * @author Jerry
 * @since 2024/10/6 17:01
 */
public class SingleLockBlockingQueue<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;
    private int tail;
    private int size;

    private ReentrantLock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();

    public SingleLockBlockingQueue(int capacity) {
        array = (E[]) new Object[capacity];
    }

    /**
     * 添加一个元素，如果队列满则阻塞
     *
     * @param e
     * @throws InterruptedException
     */
    @Override
    public void offer(E e) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (isFull()) {
                notFull.await();
            }
            array[tail++] = e;
            if (tail == array.length) {
                tail = 0;
            }
            size++;
            // 唤醒等待非空的线程
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 在指定时间内添加一个元素，如果队列满则阻塞
     *
     * @param e
     * @param timeout 毫秒数
     * @return 添加成功或者失败
     * @throws InterruptedException
     */
    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            long remainder = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()) {
                if (remainder <= 0) {
                    return false;
                }
                remainder = notFull.awaitNanos(remainder);
            }
            array[tail++] = e;
            if (tail == array.length) {
                tail = 0;
            }
            size++;
            // 唤醒等待非空的线程
            notEmpty.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 移除一个元素，如果队列空则阻塞
     *
     * @return
     * @throws InterruptedException
     */
    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly();
        E e;
        try {
            while (isEmpty()) {
                notEmpty.await();
            }
            e = array[head];
            array[head] = null;
            if (++head == array.length) {
                head = 0;
            }
            size--;
            // 唤醒等待不满的线程
            notFull.signal();
            return e;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 移除一个元素，如果队列空则阻塞
     *
     * @return
     * @throws InterruptedException
     */
    @Override
    public E poll(long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        E e;
        try {
            long remainder = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isEmpty()) {
                if (remainder <= 0) {
                    return null;
                }
                remainder = notEmpty.awaitNanos(remainder);
            }
            e = array[head];
            array[head] = null;
            if (++head == array.length) {
                head = 0;
            }
            size--;
            // 唤醒等待不满的线程
            notFull.signal();
            return e;
        } finally {
            lock.unlock();
        }
    }

    private boolean isEmpty() {
        return size == 0;
    }

    private boolean isFull() {
        return size == array.length;
    }

    @Override
    public String toString() {
        return "array=" + Arrays.toString(array) + '}';
    }
}
